package study.flink;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SocketTextStreamFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Date;
import java.util.UUID;

public class TimerTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        OutputTag<String> side = new OutputTag<String>("timer") {
        };
        SingleOutputStreamOperator<String> singleOutputStreamOperator = env.addSource(new SocketTextStreamFunction("vm-ubuntu", 8888, "\n", 1))
                .flatMap(new FlatMapFunction<String, Character>() {
                    @Override
                    public void flatMap(String value, Collector<Character> out) throws Exception {
                        for (char c : value.toCharArray()) {
                            out.collect(c);
                        }
                    }
                })
                .filter(value -> '0' <= value && '9' >= value)
                .map(new MapFunction<Character, Tuple2<Integer, Long>>() {
                    @Override
                    public Tuple2<Integer, Long> map(Character value) throws Exception {
                        return new Tuple2<>((int) value, System.currentTimeMillis());
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<Integer, Long>>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner((element, recordTimestamp) -> element.getField(1)))
                .keyBy(new KeySelector<Tuple2<Integer, Long>, Integer>() {
                    @Override
                    public Integer getKey(Tuple2<Integer, Long> value) throws Exception {
                        return value.getField(0);
                    }
                })
                .process(new KeyedProcessFunction<Integer, Tuple2<Integer, Long>, String>() {

                    ValueState<String> valueState;

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        ctx.output(side, valueState.value());
                    }

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        valueState = getRuntimeContext().getState(new ValueStateDescriptor<>("test", String.class));
                    }

                    @Override
                    public void processElement(Tuple2<Integer, Long> value, Context ctx, Collector<String> out) throws Exception {
                        String uuid = UUID.randomUUID().toString();
                        valueState.update(uuid);
                        ctx.timerService().registerProcessingTimeTimer(ctx.timestamp() + 3000);
                        out.collect(uuid);
                    }
                });
        singleOutputStreamOperator.getSideOutput(side).addSink(new SinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) throws Exception {
                System.out.println(String.format("side %s: %s", new Date(), value));
            }
        });
        singleOutputStreamOperator.addSink(new SinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) throws Exception {
                System.out.println(String.format("main %s: %s", new Date(), value));
            }
        });

        env.execute();
    }

}
